/**
* @Author  : zou.yingbin
* @Create  : 2022/1/10 15:20
* @Modify  : 2022/1/10 15:20
* @note    :
 */

package rabbitmq

import (
	"fmt"
	"github.com/streadway/amqp"
)

//ReceiveChannel 接收队列
type ReceiveChannel struct {
	topic      string        // routing key
	autoAck    bool          //
	autoDelete bool          // 通道关闭时自动删除此队列
	c          *amqp.Channel // 通道
	q          amqp.Queue    // 队列
	queueName  string        // 队列名称
	err        error
}

func (r *ReceiveChannel) init(conn *amqp.Connection,
	cfg MqConfig, f MessageCallBack) error {
	r.err = nil
	r.openChannel(conn)
	r.exchangeDeclare(cfg)
	r.queueDeclare()
	r.queueBind(cfg)
	go r.consumeMsg(f)
	return r.err
}

func (r *ReceiveChannel) openChannel(conn *amqp.Connection) {
	if r.err == nil {
		r.c, r.err = conn.Channel()
	}
}

func (r *ReceiveChannel) exchangeDeclare(cfg MqConfig) {
	if r.err == nil {
		r.err = r.c.ExchangeDeclare(
			cfg.ExchangeName, // Exchange
			cfg.ExchangeType, // Type
			false,            // Durable
			false,            // AutoDelete
			false,            // Internal
			false,            // noWait
			nil,              // Table
		)
	}
}

func (r *ReceiveChannel) queueDeclare() {
	if r.err == nil {
		r.q, r.err = r.c.QueueDeclare(
			r.queueName, // Name
			true,        // Durable
			false,       // AutoDelete
			false,       // Exclusive
			false,       // nowait
			nil,         // Table
		)
	}
}

func (r *ReceiveChannel) queueBind(cfg MqConfig) {
	if r.err == nil {
		r.err = r.c.QueueBind(
			r.q.Name,         // Name
			r.topic,          // key
			cfg.ExchangeName, // exchange
			true,             // noWait
			nil,              // Table
		)
	}
}

func (r *ReceiveChannel) consumeMsg(f MessageCallBack) {
	if r.err == nil {
		msgLst, err := r.c.Consume(
			r.q.Name,
			"ada",
			r.autoAck,
			false,
			false,
			false,
			nil,
		)
		r.err = err
		for i := 0; i < 4; i++ {
			go func(work <-chan amqp.Delivery) {
				for msg := range work {
					f(msg.Body, msg.DeliveryTag, r)
				}
				fmt.Println("consume exit")
			}(msgLst)
		}
	}
}

//Ack 消息, 自动确认模式, 不需要确认, 否则调用确认接口
func (r *ReceiveChannel) Ack(deliveryTag uint64) error {
	if r.autoAck {
		return nil
	}
	err := r.c.Ack(deliveryTag, false)
	if err != nil {
		fmt.Println("ack error:", err)
	}
	return nil
}

func (r *ReceiveChannel) deleteQueue() {
	if r.c != nil {
		_, _ = r.c.QueueDelete(
			r.queueName,
			false,
			false,
			true)
		r.c.Close()
	}
}

//SenderChannel 发送队列
type SenderChannel struct {
	c *amqp.Channel // 通道
}

func (r *SenderChannel) init(conn *amqp.Connection, cfg MqConfig) error {
	var err error
	r.c, err = conn.Channel()
	if err == nil {
		r.c.ExchangeDeclare(
			cfg.ExchangeName,
			cfg.ExchangeType,
			false, false, false, false, nil)
	}
	return err
}
